package org.example.app.dwd;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.example.utils.DruidDSUtil;
import org.example.utils.MyKafkaUtil;
import java.sql.Connection;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * 搜索关键词事实表处理器
 */
public class DwdMarketSearchProcessor {
    
    public static void main(String[] args) throws Exception {
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "ods_market_search";
        String groupId = "dwd_market_search_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 添加水位线
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(jsonStr -> {
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
            return jsonObject;
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                })
        );
        
        // 将结果写入Kafka
        jsonObjDS.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject jsonObject) throws Exception {
                return jsonObject.toJSONString();
            }
        }).addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_market_search"));
        
        // 执行任务
        env.execute("Search Keyword Fact Table Processing Job");
    }
    
    // 用于获取数据库连接的工具方法
    private static Connection getConnection() throws Exception {
        return DruidDSUtil.createDataSource().getConnection();
    }
}